package com.fanxl.flink.datastream.source;

import com.fanxl.flink.datastream.transformation.Access;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;

import java.util.Random;

/**
 * @description:
 * @author: fanxiaole
 * @date: 2022/1/13 10:59
 */
public class AccessSourceV2 implements ParallelSourceFunction<Access> {

    private boolean running = true;

    @Override
    public void run(SourceContext<Access> sourceContext) throws Exception {

        Random random = new Random();

        String[] domains = {"a.com", "b.com", "baidu.com"};

        while (running) {
            for (int i = 0; i < 8; i++) {
                sourceContext.collect(new Access(123455L, domains[random.nextInt(domains.length)], random.nextDouble() + 1000));
            }
            Thread.sleep(5000);
        }
    }

    @Override
    public void cancel() {
        running = false;
    }
}
